package com.zhang.flink.example.sink;

import com.zhang.flink.bean.ClickEvent;
import com.zhang.flink.example.source.ClickSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @title: 自定义sink 实现幂等写入MySQL
 * @author: zhang
 * @date: 2022/2/11 20:31
 */
public class MysqlSink {
    public static void main(String[] args) throws Exception {
        // todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .addSink(new SinkToMySQL());

        // todo 启动任务
        env.execute();
    }

    public static class SinkToMySQL extends RichSinkFunction<ClickEvent> {
        private Connection connection;
        private PreparedStatement insertStmt;
        private PreparedStatement updateStmt;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection(
                    "jdbc:mysql://hadoop102:3306/userbehavior?useSSL=false",
                    "root",
                    "000000"
            );
            insertStmt = connection.prepareStatement("insert into clicks (name,url) values (?,?)");
            updateStmt = connection.prepareStatement("update clicks set url = ? where name = ?");
        }

        @Override
        public void invoke(ClickEvent value, Context context) throws Exception {
            updateStmt.setString(1, value.url);
            updateStmt.setString(2, value.name);
            updateStmt.execute();
            if (updateStmt.getUpdateCount()==0){
                insertStmt.setString(1, value.name);
                insertStmt.setString(2, value.url);
                insertStmt.execute();
            }
        }

        @Override
        public void close() throws Exception {
            insertStmt.close();
            updateStmt.close();
            connection.close();
        }
    }
}
